package com.yichat.micro.msghttpservice.core.record.batch;

import com.weibo.api.motan.util.ConcurrentHashSet;
import com.yichat.micro.msghttpservice.common.util.JsonUtil;
import com.yichat.micro.msghttpservice.core.record.mapper.RecordMapper;
import com.yichat.micro.msghttpservice.core.record.model.Record;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.stream.Collectors;

/**
 * Created by chenmohan on 2019/04/25
 *
 * @description 单次请求的批量处理
 */
@Slf4j
@Component
public class RecordBatch {

    @Value("${msg-http-service.record.batch-insert.capacity}")
    Integer insertCapacity;

    @Autowired
    RecordMapper recordMapper;

    protected BlockingQueue<Record> insertQueue;

    protected BlockingQueue<Record> updateQueue;

    @PostConstruct
    public void init() {
        insertQueue = new ArrayBlockingQueue<>(insertCapacity);
        updateQueue = new ArrayBlockingQueue<>(insertCapacity);
    }

    /**
     * 插入中的record对象id,在批量入库后会删除对应的id
     */
    protected ConcurrentHashSet<Long> insertingRecordSet = new ConcurrentHashSet<>();

    public void insert(Record record) {
        insertQueue.add(record);
        insertingRecordSet.add(record.getId());
    }

    public void update(Record record) {
        updateQueue.add(record);
    }

    @Scheduled(cron = "0/5 * *  * * ? ")
    public void batch() {
        insert();
        //每次新增请求日志完成后再执行更新请求日志操作
        update();
    }


    public void insert() {
        List<Record> batchInsertList = new ArrayList<>(100);
        Set<Long> removingIdSet = new HashSet<>(100);
        //一次最多写入5000条请求日志
        insertQueue.drainTo(batchInsertList, 5000);
        if (batchInsertList.size() == 0) {
            return;
        }
        try {
            recordMapper.batchInsert(batchInsertList);
            batchInsertList.parallelStream().map(e -> e.getId()).collect(Collectors.toSet());
            List<Long> recordIds = batchInsertList.parallelStream().map(e -> e.getId()).collect(Collectors.toList());
            log.info("id={}入库", Arrays.toString(recordIds.toArray()));
            insertingRecordSet.removeAll(recordIds);
        } catch (Exception e) {
            log.error("批量插入Record异常", e);
            for (Record r : batchInsertList) {
                try {
                    recordMapper.insertSelective(r);
                    removingIdSet.add(r.getId());
                } catch (Exception e2) {
                    log.error("单条请求插入Record异常 , info => {}", JsonUtil.toJSON(r), e2);
                }
            }
        }
        //删除
        insertingRecordSet.removeAll(removingIdSet);
    }

    public void update() {
        List<Record> updateList = new ArrayList<>(100);
        List<Record> updateFailedList = new ArrayList<>(100);
        updateQueue.drainTo(updateList, 500);
        if (updateList.size() == 0) {
            return;
        }
        for (Record r : updateList) {
            if (!insertingRecordSet.contains(r.getId())) {  //不在待插入列表中,说明已经入库
                int res = recordMapper.updateByPrimaryKeySelective(r);
        //        log.info("id={}还没有入库,不能更新", r.getId());
                if (res == 0) {//等于0表示数据库中没有该条日志,更新失败
                    updateFailedList.add(r);
                }
            } else {  //在的话,还没有入库,不能先更新
         //       log.info("id={}还没有入库,不能更新", r.getId());
                updateFailedList.add(r);
            }
        }
        if (updateFailedList.size() > 0) {      //如果有更新失败的日志,则放回到待更新日志
       //     log.warn("存在更新失败的请求日志: " + Arrays.toString(updateFailedList.parallelStream().map(e -> e.getId()).collect(Collectors.toList()).toArray()));
            updateQueue.addAll(updateFailedList);
        }
    }
}
